/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.common;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.pleiades.client.LockServiceClient;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.persistence.LockStatus;
import com.nhncorp.pleiades.protocol.LockMetaData;

/**
 * pleiades에서 persistent한 lock에 대한 기능을 제공하지 않기 때문에 임시로 lock 관련 정보를
 * DFS에 저장하도록 구현
 * pleiades에서 persistent한 lock을 구현하면 pleiades의 기능을 사용하고 이 클래스는 삭제
 * @author jindolk
 *
 */
public class NeptuneLockService extends LockServiceClient {
  public static final Log LOG = LogFactory.getLog(NeptuneLockService.class.getName());
  
  static final String META_FILE = "LOCK.META";
  NConfiguration nconf;
  NeptuneFileSystem fs;
  GPath rootPath;
  
  public NeptuneLockService(PleiadesUser user, PleiadesConfiguration conf, NConfiguration nconf) throws IOException {
    super(user, conf);
    this.nconf = nconf;
    this.fs = NeptuneFileSystem.get(nconf);
    this.rootPath = new GPath(nconf.get("neptune.root"));
  }

  @Override
  public boolean createNode(LockMetaData metaData, boolean overWrite,
      boolean isDirectory) throws IOException {
    boolean result = super.createNode(metaData, overWrite, isDirectory);
    if(!result) {
      return result;
    }
    GPath path = new GPath(rootPath + metaData.getLockId());
    
    if(overWrite && fs.exists(path)) {
      fs.delete(path, true);
    }
    
    boolean success = fs.mkdirs(path.getParent());
    if(!success) {
      return false;
    }
    
    DataOutputStream out = null;
    try {
      GPath metaFilePath = new GPath(path, META_FILE);
      
      //LOG.fatal("metaFilePath:" + metaFilePath + ",rootPath=" + rootPath + ",path=" + path + ",lockId=" + metaData.getLockId());
      out = new DataOutputStream(fs.create(metaFilePath));
      metaData.write(out);
    } finally {
      if(out != null) {
        out.close();
      }
    }
    return true;
  }

  @Override
  public boolean deleteNode(String path, boolean recursive) throws IOException {
    boolean result = super.deleteNode(path, recursive);
    if(!result) {
      //lock이 존재하는 경우에는 그냥 false 반환
      if(getLockStatus(path) != null) {
        return result;
      }
      return result;
    }
    GPath gpath = new GPath(rootPath + addServiceId(serviceId, path));
    boolean success = fs.delete(gpath, recursive);
    if(!success) {
      return success;
    }
    GPath metaFilePath = new GPath(gpath, META_FILE);
    if(fs.exists(metaFilePath)) {
      return fs.delete(new GPath(gpath, META_FILE));
    } else {
      return true;
    }
  }

  @Override
  public boolean exist(String path) throws IOException {
    GPath gpath = new GPath(rootPath + addServiceId(serviceId, path));
    boolean result = fs.exists(gpath);
    //LOG.warn("exists:" + path + "," + gpath + "," + result);
    return result;
  }

  @Override
  public LockMetaData getMetaData(String path) throws IOException {
    LockMetaData lockMetaData = super.getMetaData(path);
    if(lockMetaData != null) {
      return lockMetaData;
    }
    GPath gpath = new GPath(rootPath + addServiceId(serviceId, path));

    //lock meta가 없으면 lock 정보를 가져와서 lock이 존재하는 지 확인한다.
    LockStatus lockStatus = super.getLockStatus(path);
    if(lockStatus == null) {
      //모두 삭제
      fs.delete(gpath, true);
      return null;
    }
    return getMetaDataFromFS(gpath);
  }

  /**
   * @param path
   * @return
   * @throws IOException
   */
  private LockMetaData getMetaDataFromFS(GPath path) throws IOException {
    LockMetaData lockMetaData;

    GPath metaFilePath = new GPath(path, META_FILE);
    if(fs.exists(metaFilePath)) {
      DataInputStream in = null;
      try {
        in = new DataInputStream(fs.open(metaFilePath));
        lockMetaData = new LockMetaData();
        lockMetaData.readFields(in);
        return lockMetaData;
      } finally {
        if(in != null) {
          in.close();
        }
      }
    } else {
      return null;
    }
  }

  public LockMetaData[] getMetaDataList(String path, boolean includeDirectory)
      throws IOException {
    LockMetaData[] lockMedaDatas = super.getMetaDataList(path, includeDirectory);
    if(lockMedaDatas != null) {
      return lockMedaDatas;
    }
    
    GPath gpath = new GPath(rootPath + addServiceId(serviceId, path));
    if(!fs.exists(gpath)) {
      return null;
    }
    GPath[] paths = fs.list(gpath);
    if(paths == null || paths.length == 0) {
      return null;
    }
    
    List<LockMetaData> result = new ArrayList<LockMetaData>(); 
    for(GPath eachPath: paths) {
      LockMetaData lockMetaData = getMetaDataFromFS(eachPath);
      if(lockMetaData != null) {
        result.add(lockMetaData);
      }
    }
    
    return result.toArray(new LockMetaData[]{});
  }

  @Override
  public Map<String, LockMetaData> readDir(String path, boolean includeDirectory)
      throws IOException {
    Map<String, LockMetaData> lockMetaDatas = super.readDir(path, includeDirectory);
    if(lockMetaDatas != null) {
      return lockMetaDatas;
    }
    
    GPath gpath = new GPath(rootPath + addServiceId(serviceId, path));
    if(!fs.exists(gpath)) {
      return null;
    }
    GPath[] paths = fs.list(gpath);
    if(paths == null || paths.length == 0) {
      return null;
    }
    
    Map<String, LockMetaData> result = new HashMap<String, LockMetaData>(); 
    for(GPath eachPath: paths) {
      LockMetaData lockMetaData = getMetaDataFromFS(eachPath);
      if(lockMetaData != null) {
        result.put(lockMetaData.getLockId(), lockMetaData);
      }
    }
    
    return result;
  }
}
